package com.oracle.itemcf;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

/**
 * 结果得到评分矩阵：
 * 	1	A_2.0	C_5.0
 * 	2	A_10.0	B_3.0
 * 	3	C_15.0
 * 	4	A_3.0	C_5.0
 * 	5	B_3.0
 * 	6	A_5.0	B_5.0
 * */
public class Job1 {
	public static class Map extends Mapper<LongWritable, Text, Text, Text> {
		
		@Override
		protected void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			String line = value.toString();
			String split[] = line.split("\t");
			
			String user = split[0];
			String item = split[1];
			String val = split[2];
			
			context.write(new Text(item), new Text(user+"_"+val));
		}
	}
	
	public static class Reduce extends Reducer<Text, Text, Text, Text> {
		
		
		@Override
		protected void reduce(Text key, Iterable<Text> values, Context context)
				throws IOException, InterruptedException {
			HashMap<String, Double> map = new HashMap<>();
			List<String> list = new ArrayList<>();
			
			String item = key.toString();
			String user;
			Double val;
			String user_item;
			
			for(Text i: values) {
				user = i.toString().substring(0, i.toString().indexOf("_"));
				val = Double.parseDouble(i.toString().substring(i.toString().indexOf("_")+1));
				user_item = user+"_"+item;
				if(!map.containsKey(user)) {
					map.put(user, val);
					list.add(user);
				}else {
					map.put(user, val+map.get(user));
				}
			}
			
			String line = "";
			for(String i: list) {
				line = line + i + "_" + map.get(i) + "\t";
			}
			context.write(key, new Text(line));
		}
	}
	
	public static void main(String[] args) throws Exception {
		Configuration conf = new Configuration();
		
		Job job = Job.getInstance(conf);
		job.setJarByClass(Job1.class);
		job.setMapperClass(Map.class);
		job.setReducerClass(Reduce.class);
		
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(Text.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);
		
		FileInputFormat.setInputPaths(job, new Path("/itemcf/input"));
		FileOutputFormat.setOutputPath(job, new Path("/itemcf/output1"));
		
		job.waitForCompletion(true);
	}
}
